package site.icesky.springboot.rocketmq.config;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import site.icesky.springboot.rocketmq.mqconsumer.AbstractRocketMqConsumer;
import site.icesky.springboot.rocketmq.mqconsumer.RocketMqConsumer;

import javax.annotation.PostConstruct;
import java.util.Iterator;
import java.util.Map;

/**
 * @ClassName: RocketMqConfig <br>
 * @Description: rocketmq的配置启动注册类 <br>
 * @Package: site.icesky.springboot.rocketmq.config <br>
 * @author: suitianmou <br>
 * @Date: 2020/5/21-20:11 <br>
 * @version: 1.0 <br>
 * @see <br>
 * @since (从产品的哪一个版本此类被添加进来 ( 可选))<br>
 * 修改历史: <br>
 * 修改时间             修改人           版本变更           修改说明<br>
 * -----------------------------------------------------<br>
 * <br>
 */
@Configuration
// 配置参数类，使用该类中的默认参数
@EnableConfigurationProperties(RocketMqProperties.class)
// prefix是配置文件的前缀,value是配置的值,合起来是rocketmq.isEnable配置参数, havingValue=true意思是如果是true才注册该类，否则不启动
@ConditionalOnProperty( prefix = "rocketmq", value = "isEnable", havingValue = "true")
public class RocketMqConfig {
    private static final Logger logger = LoggerFactory.getLogger(RocketMqConfig.class);

    @Autowired
    RocketMqProperties rocketMqProperties;

    @Autowired
    private ApplicationContext applicationContext;

    /**
     * @Title:  RocketMqConfig  <br>
     * @Description:  构造方法  <br>
     * @version:  1.0  <br>
     * @param:  [rocketMqProperties, applicationContext]  <br>
     * @return:    <br>
     * @author:  suitianmou  <br>
     * @Date:  2020/5/21-20:27  <br>
     * history:  <br>
     * 修改记录：  修改人           版本变更           修改说明<br>
     * -----------------------------------------------------<br>
     * <br>
     */
    public RocketMqConfig(RocketMqProperties rocketMqProperties, ApplicationContext applicationContext){
        this.applicationContext = applicationContext;
        this.rocketMqProperties = rocketMqProperties;
    }
    /**
     * @Title:  getMqProducter  <br>
     * @Description:  生产者的注册方法  <br>
     * @version:  1.0  <br>
     * @param:  []  <br>
     * @return:  org.apache.rocketmq.client.producer.DefaultMQProducer  <br>
     * @author:  suitianmou  <br>
     * @Date:  2020/5/23-19:42  <br>
     * history:  <br>
     * 修改记录：  修改人           版本变更           修改说明<br>
     * -----------------------------------------------------<br>
     * <br>
     */
    @Bean
    public DefaultMQProducer getMqProducter() throws MQClientException{

        System.setProperty("rocketmq.client.logUseSlf4j", "true");

        // 构造MQ生产者
        DefaultMQProducer defaultMQProducer = new DefaultMQProducer(rocketMqProperties.getGroupName());

        if( rocketMqProperties.getNamesvrAddr() == null || "".equals(rocketMqProperties.getNamesvrAddr())){
            System.out.println("MQ NAMESERVER 必须配置");
            throw new MQClientException( -1, "NAMESERVER 不能为空");
        }

        // 赋值参数
        defaultMQProducer.setNamesrvAddr( rocketMqProperties.getNamesvrAddr());
        defaultMQProducer.setVipChannelEnabled(false);
        defaultMQProducer.setMaxMessageSize(rocketMqProperties.getProoductMaxMessageSize());
        defaultMQProducer.setRetryTimesWhenSendAsyncFailed(rocketMqProperties.getProductRetryTimesWhenSndFail());
        defaultMQProducer.setSendMsgTimeout(rocketMqProperties.getProductSendMessageTimeOut());
        defaultMQProducer.setCompressMsgBodyOverHowmuch(rocketMqProperties.getCompressMsgBodyOverHowmuch());

        // MQ启动生产者
        defaultMQProducer.start();
        return defaultMQProducer;
    }

    // 比bean先执行,类似前处理，但是同样的前处理无顺序
    @PostConstruct
    public void initConsumser() throws MQClientException{

        System.setProperty("rocketmq.client.logUseSlf4j", "true");

        /*1. 获取所有MQ消费者对象，可以有多个，下面的方法会查到所有继承了AbstrackRocketMqConsumer的类 */
        Map<String, AbstractRocketMqConsumer> consumerMap =
                applicationContext.getBeansOfType(AbstractRocketMqConsumer.class);
        if( consumerMap == null || consumerMap.size() == 0){
            logger.warn("没有注册的消费者");
            return;
        }

        /*2. 循环启动所有消费者 */
        Iterator<String> beans = consumerMap.keySet().iterator();
        while( beans.hasNext()){
            String beanName = (String) beans.next();
            AbstractRocketMqConsumer consumer = consumerMap.get(beanName);

            /*初始化消费者 */
            consumer.init();

            //logger 启动消费者

            /*创建消费者*/
            createConsumer( consumer);
        }
    }

    public void createConsumer( AbstractRocketMqConsumer arc) throws MQClientException{

        /* 1.创建消费者对象 */
        DefaultMQPushConsumer consumer  = new DefaultMQPushConsumer();

        /* 2.设置消费者参数 */
        consumer.setNamesrvAddr(rocketMqProperties.getNamesvrAddr());
        consumer.setConsumerGroup(rocketMqProperties.getGroupName());
        consumer.setConsumeThreadMin(rocketMqProperties.getConsumerThreadMin());
        consumer.setConsumeThreadMax(rocketMqProperties.getConsumerThreadMax());
        consumer.setPullInterval(rocketMqProperties.getPullInterval());

        consumer.registerMessageListener(arc.getMessageListenerConcurrently());
        consumer.setConsumeMessageBatchMaxSize(rocketMqProperties.getConsumerMessageBathMaxSize());
        consumer.setMessageModel(arc.getMessageModel());
        consumer.setMaxReconsumeTimes(rocketMqProperties.getMaxReconsumeTime());
        consumer.setPullBatchSize(rocketMqProperties.getPullBatchSize());


        try{
            /* 3.订阅队列 */
            consumer.subscribe( arc.getTopics(), arc.getTags());

            /* 4.启动消费者 */
            consumer.start();
            //RocketmqMqShutdownElement.addConsume(consumer);

            arc.setMqPushConsumer(consumer);
        }catch(MQClientException e){
            // 监听启动异常
            throw e;
        }
    }
}
